package com.ophiux.batch;

import com.alibaba.fastjson.JSONObject;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * @desc: 批量消息
 * @author: hhl
 * @date：2023/8/16 13:54
 */
public class BatchProducer {

    public static void main(String[] args) throws MQClientException, MQBrokerException, RemotingException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("BatchProducer");
        producer.setNamesrvAddr("127.0.0.1:9876");
        producer.start();

        List<Message> list = new ArrayList<>();
        for (int i = 0; i < 100000; i++) {
            Message message = new Message("Simple", "Tags", (i + "_BatchProducer").getBytes(StandardCharsets.UTF_8));
            list.add(message);
        }
        //批量发送
//        SendResult send = producer.send(list);
//        System.out.println(send);

        //分批发送批量消息
        ListSplitter listSplitter = new ListSplitter(list);
        int i = 0;
        while (listSplitter.hasNext()) {
            SendResult send = producer.send(listSplitter.next());
            System.out.println(i++ + "消息发送成功" + send);
        }
        producer.shutdown();
    }
}

class ListSplitter implements Iterator<List<Message>> {

    private static final int SIZE_LIMIT = 40 * 1000;
    private final List<Message> messages;

    private int currentIndex = 0;

    ListSplitter(List<Message> messages) {
        this.messages = messages;
    }

    @Override
    public boolean hasNext() {
        return currentIndex < messages.size();
    }

    @Override
    public List<Message> next() {
        int nextIndex = currentIndex;
        int totalSize = 0;
        for (; nextIndex < messages.size(); nextIndex++) {
            Message message = messages.get(nextIndex);
            int messageSize = message.getBody().length + message.getTopic().length();
            Map<String, String> properties = message.getProperties();
            Iterator<Map.Entry<String, String>> iterator = properties.entrySet().iterator();
            while (iterator.hasNext()) {
                messageSize += iterator.next().getKey().length() + iterator.next().getValue().length();
            }
            messageSize = messageSize + 20;
            if (messageSize > SIZE_LIMIT) {
                //第一次发送就超出了限制，直接跳过这条消息继续扫描
                if (nextIndex - currentIndex == 0) {
                    nextIndex++;
                }
                break;
            }
            //如果当前发送列表已经超出了限制，暂停添加消息
            if (messageSize + totalSize > SIZE_LIMIT) {
                break;
            } else {
                totalSize += messageSize;
            }
        }
        List<Message> messages = this.messages.subList(currentIndex, nextIndex);
        currentIndex = nextIndex;
        return messages;
    }
}